package org.jxkj.app

import org.apache.spark.SparkContext
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.{SparkSession, types}
import org.jxkj.data.IDaoProvider

import java.text.SimpleDateFormat
import java.util.Date

/**
 * 变损率及厂用电计算
 */
object TrfLoss {
  /**
   * 智能抄表调用该方法
   *
   * @param ss
   * @param sc
   * @param hTable
   * @param time
   * @param step
   */
  def transformData(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int): Unit = {
    val startTime = time(0).substring(0, 10)
    val endTime = time(1).substring(0, 10)

    // 读mysql
    val hms_stat_caldata = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_stat_caldata")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    hms_stat_caldata.createOrReplaceTempView("hms_stat_caldata")

    //
    val etldf = ss.sql(
      s"""
         | select *
         | from
         |   hms_stat_caldata
         | where
         |   sub_org_id = 'formalu'
         |   and stat_type in ('day', 'month', 'year')
         |   and substr(stat_date, 1, 10) between '$startTime' and '$endTime'
         |""".stripMargin).cache()
    etldf.createOrReplaceTempView("t_etl")


    // 1、计算变损 ======================================================================================================
    val resultDf = ss.sql(
      s"""
         | select
         |   (case when stat_type='day' then '3' when stat_type='month' then '2' when stat_type='year' then '1' end) as type,
         |   stat_date as time,
         |   max(case when point_code = 'HMS_DLFX_DJ_JZDL' then calvalue end) as daj_energy,
         |   max(case when point_code = 'HMS_DLFX_EJ_JZDL' then calvalue end) as erj_energy,
         |   max(case when point_code = 'HMS_DLFX_DJ_GKSDL' then calvalue end) as daj_grid_energy,
         |   max(case when point_code = 'HMS_DLFX_EJ_GKSDL' then calvalue end) as erj_grid_energy,
         |   max(case when point_code = 'HMS_DLFX_DJ_GYBDL' then calvalue end) as daj_trf_energy,
         |   max(case when point_code = 'HMS_DLFX_EJ_GYBDL' then calvalue end) as erj_trf_energy,
         |   max(case when point_code = 'HMS_DLFX_EJ_ZYBDL' then calvalue end) as erj_trf_loss,
         |   max(case when point_code = 'HMS_DLFX_EJ_QTYDDL' then calvalue end) as qt_trf_loss,
         |   max(case when point_code = 'HMS_DLFX_DJ_BSL' then calvalue end) as rate_trf_daj,
         |   max(case when point_code = 'HMS_DLFX_EJ_BSL' then calvalue end) as rate_trf_erj,
         |   max(case when point_code = 'HMS_DLFX_QC_BSL' then calvalue end) as rate_trf_all,
         |   'GZB' as project_id,
         |   '10000' as org_id,
         |   '' as sub_org_id,
         |   '' as create_person,
         |   now() as create_time,
         |   '' as modify_person,
         |   now() as modify_time
         | from
         |   t_etl
         | group by
         |   (case when stat_type='day' then '3' when stat_type='month' then '2' when stat_type='year' then '1' end),
         |   stat_date
         |""".stripMargin)

    // 清除旧数据
    val tablename2 = "hms_eo_trfloss_analysis"
    val delString2 = s"delete from $tablename2 where substr(time,1,10) between '$startTime' and '$endTime' and type in ('1','2','3')"
    MysqlDml.delete2(delString2)

    // 结果写入数据库
    resultDf.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tablename2)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()


    // 1、计算厂用电 ====================================================================================================
    val resultDf2 = ss.sql(
      s"""
         | select
         |   stat_date time,
         |   (case when stat_type='day' then '3' when stat_type='month' then '2' when stat_type='year' then '1' end) as type,
         |   max(case when point_code='HMS_DLFX_EJ_JZDL' then calvalue end) as erj_energy,
         |   max(case when point_code='HMS_DLFX_EJ_CYDDL' then calvalue end) as erj_fs_energy,
         |   max(case when point_code='HMS_DLFX_DJ_JZDL' then calvalue end) as daj_energy,
         |   max(case when point_code='HMS_DLFX_DJ_CYDDL' then calvalue end) as daj_fs_energy,
         |   max(case when point_code='HMS_DLFX_QC_FDL' then calvalue end) as all_energy,
         |   max(case when point_code='HMS_DLFX_QC_CYDDL' then calvalue end) as all_fs_energy,
         |   max(case when point_code='HMS_DLFX_QC_CYDL' then calvalue end) as rate_fs,
         |   'GZB' as project_id,
         |   '10000' as org_id,
         |   '' as sub_org_id,
         |   '' as create_person,
         |   now() as create_time,
         |   '' as modify_person,
         |   now() as modify_time
         | from
         |   t_etl
         | group by
         |   (case when stat_type='day' then '3' when stat_type='month' then '2' when stat_type='year' then '1' end),
         |   stat_date
         |""".stripMargin)

    // 清除旧数据
    val tablename = "hms_eo_fs_analysis"
    val delString = s"delete from $tablename where substr(time,1,10) between '$startTime' and '$endTime'  and type in ('1','2','3')"
    MysqlDml.delete2(delString)

    // 结果写入数据库
    resultDf2.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tablename)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()
  }


  /**
   * 变损率厂用电-3
   *
   * @param ss
   * @param sc
   * @param hTable
   * @param time
   * @param step
   */
  def calcTransformLoss(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int): Unit = {
    import ss.implicits._
    var startTime = time(0)
    val lastDate = time(1)
    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
    //    var endTimeStamp = sdf.parse(lastDate).getTime()
    //    var endTime = sdf.format(new Date(endTimeStamp + 3600 * 24 * 1000))

    var startTime2 = startTime //startTime + " 00:00:00"
    val endTime2 = lastDate //endTime + " 23:59:59"
    val time2 = Array(startTime2, endTime2)

    // 读取配置文件
    val schema = new StructType()
      .add("point_name", types.StringType)
      .add("point_code", types.StringType)
      .add("tlc_branch", types.StringType)
      .add("multi_ratio", types.DoubleType)
    val hms_eo_trfloss_config_DF = ss.read.schema(schema).csv(MysqlDml.fileprefix + "/user/hive/trfloss.csv")
    hms_eo_trfloss_config_DF.createOrReplaceTempView("hms_eo_trfloss_config")

    // 测点集合
    val point_code: Array[String] = hms_eo_trfloss_config_DF.map(x => x.getString(1)).collect()

    // 读取配置表
    ss.sql(
      s"""
         | select
         |   point_name,
         |   point_code,
         |   tlc_branch,
         |   multi_ratio as org_id,
         |   (case when instr(point_name,'励磁') > 0 or instr(point_code,'P2R0') > 0 then -1 else 1 end) as is_positive
         | from
         |   hms_eo_trfloss_config
         |""".stripMargin)
      .createOrReplaceTempView("hms_eo_trfloss_config2")

    // 获取测点的采集数据
    GetData.getByWebApi(ss, sc, hTable, point_code, time2, step, 900)

    val etlDF15 = ss.sql(
      s"""
         | select
         |   a.checkPoint,
         |   b.createTime,
         |   a.pvalue-b.pvalue as pvalue
         | from
         |   t_ods_hbase_edos_tmp a join t_ods_hbase_edos_tmp b on
         |     a.checkPoint = b.checkPoint and unix_timestamp(a.createTime, 'yyyy/MM/dd HH:mm:ss') = unix_timestamp(b.createTime, 'yyyy/MM/dd HH:mm:ss')+60*15
         | where
         |   b.createTime between '$startTime' and '$lastDate'
         |""".stripMargin)
    etlDF15.createOrReplaceTempView("t_ods_hbase_edos15")


    /* 变损电量逻辑
    * daj_energy --大江发电量
    * daj_grid_energy --大江 500kv上网电量
    * daj_trf_energy --大江厂用变电量
    * erj_energy --二江发电量
    * erj_grid_energy --二江 220kv上网电量
    * erj_trf_energy --二江厂用变电量
    * erj_trf_loss --二江自用变电量
    * qt_trf_loss --其他损耗电量
    * daj_cyd --大江...
    * erj_cyd --二江...
    * */

    val tmp1resultDf15 = ss.sql(
      s"""
         | select
         |   createTime,
         |   sum(case when tlc_branch = 'daj_energy' then pvalue*org_id else 0 end)/100 as daj_energy,
         |   sum(case when tlc_branch = 'daj_grid_energy' then pvalue*org_id*is_positive else 0 end)/10000 as daj_grid_energy,
         |   sum(case when tlc_branch = 'daj_trf_energy' then pvalue*org_id else 0 end)/100 as daj_trf_energy,
         |   sum(case when tlc_branch = 'erj_energy' then pvalue*org_id*is_positive else 0 end)/100 as erj_energy,
         |   sum(case when tlc_branch = 'erj_grid_energy' then pvalue*org_id*is_positive else 0 end)/10000 as erj_grid_energy,
         |   sum(case when tlc_branch = 'erj_trf_energy' then pvalue*org_id else 0 end)/1000 as erj_trf_energy,
         |   sum(case when tlc_branch = 'erj_self_energy' then pvalue*org_id else 0 end)/100 as erj_trf_loss,
         |   sum(case when tlc_branch = 'erj_other_energy' then pvalue*org_id else 0 end)/100 as qt_trf_loss,
         |   sum(case when tlc_branch = 'daj_cyd' then pvalue*org_id else 0 end)/100 as daj_cyd,
         |   sum(case when tlc_branch = 'erj_cyd' then pvalue*org_id else 0 end)/100 as erj_cyd
         | from
         |   t_ods_hbase_edos15 a left join hms_eo_trfloss_config2 b on
         |     a.checkPoint = b.point_code
         | group by
         |   createTime
         |""".stripMargin).cache()
    ss.catalog.uncacheTable("t_ods_hbase_edos_tmp")
    tmp1resultDf15.createOrReplaceTempView("tmp1resultDf15")


    /**
     * 变损率 15分报表（计算结果）
     * rate_trf_daj --大江变损率
     * rate_trf_erj --二江变损率
     * rate_trf_all --全厂变损率
     */
    val resultDf = ss.sql(
      s"""
         | select
         |   '5' as type,
         |   createTime as time,
         |   daj_energy,
         |   erj_energy,
         |   daj_grid_energy,
         |   erj_grid_energy,
         |   daj_trf_energy,
         |   erj_trf_energy,
         |   erj_trf_loss,
         |   qt_trf_loss,
         |   (daj_energy-daj_grid_energy-daj_trf_energy)/daj_energy as rate_trf_daj,
         |   (erj_energy-erj_grid_energy-erj_trf_energy-erj_trf_loss-qt_trf_loss)/erj_energy as rate_trf_erj,
         |   (daj_energy+erj_energy-daj_grid_energy-daj_trf_energy-erj_grid_energy-erj_trf_energy-erj_trf_loss-qt_trf_loss)/(daj_energy+erj_energy) as rate_trf_all,
         |   'GZB' as project_id,
         |   '10000' as org_id,
         |   '' as sub_org_id,
         |   '' as create_person,
         |   now() as create_time,
         |   '' as modify_person,
         |   now() as modify_time
         | from
         |   tmp1resultDf15
         | where
         |   daj_energy <> 0
         |""".stripMargin)

    // 清理旧数据
    val tablename = "hms_eo_trfloss_analysis"
    val delString = s"delete from $tablename where time between '$startTime' and '$lastDate' and type='5'"
    MysqlDml.delete2(delString)

    // 结果写入数据库
    resultDf.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tablename)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()

    ///////变损率 日报表substr(createTime,12,8)='00:00:00   ？？？？代码呢？


    /**
     * 厂用电 15分报表（计算结果）
     * erj_energy --二江发电量
     * erj_fs_energy --二江厂用电量
     * daj_energy --大江发电量
     * daj_fs_energy --大江厂用电量
     * all_energy --全厂发电量
     * all_fs_energy --全厂用电量
     * rate_fs --全厂用电率
     */
    val result2Df15 = ss.sql(
      s"""
         | select
         |   '5' as type,
         |   createTime as time,
         |   erj_energy,
         |   (erj_cyd+erj_trf_loss) as erj_fs_energy,
         |   daj_energy,
         |   daj_cyd as daj_fs_energy,
         |   (erj_energy+daj_energy) as all_energy,
         |   (erj_cyd+erj_trf_loss+daj_cyd) as all_fs_energy,
         |   (erj_cyd+erj_trf_loss+daj_cyd)/(daj_energy+erj_energy) as rate_fs,
         |   'GZB' as project_id,
         |   '10000' as org_id,
         |   '' as sub_org_id,
         |   '' as create_person,
         |   now() as create_time,
         |   '' as modify_person,
         |   now() as modify_time
         | from
         |   tmp1resultDf15
         | where
         |   daj_energy <> 0
         |""".stripMargin)

    // 清理旧数据
    val tablename2 = "hms_eo_fs_analysis"
    val delString2_15 = s"delete from $tablename2 where time between '$startTime' and '$lastDate' and type='5'"
    MysqlDml.delete2(delString2_15)

    // 结果写入数据库
    result2Df15.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tablename2)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()
  }


  //@202107:未使用
  @deprecated
  def calculateData_mth(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int): Unit = {
    import ss.implicits._
    var startTime = time(0)
    val lastDate = time(1)
    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
    var endTimeStamp = sdf.parse(lastDate).getTime()
    var endTime = sdf.format(new Date(endTimeStamp + 3600 * 24 * 1000))
    println(endTime)
    println(endTime.substring(8, 10) + "a")
    var startTime2 = lastDate.substring(0, 8) + "01 00:00:00"
    val endTime2 = endTime + " 00:00:00"
    val time1 = Array(startTime2, startTime2)
    val time2 = Array(endTime2, endTime2)
    if (endTime.substring(8, 10) == "01") {
      val schema = new StructType()
        .add("point_name", types.StringType)
        .add("point_code", types.StringType)
        .add("tlc_branch", types.StringType)
        .add("multi_ratio", types.DoubleType)

      val hms_eo_trfloss_config_DF = ss.read.schema(schema).csv(MysqlDml.fileprefix + "/user/hive/trfloss.csv")
      hms_eo_trfloss_config_DF.createOrReplaceTempView("hms_eo_trfloss_config")

      ss.sql("select point_name,point_code,tlc_branch,multi_ratio org_id,case when instr(point_name,'励磁')>0 or instr(point_code,'P2R0')>0 then -1 else 1 end as is_positive from hms_eo_trfloss_config").createOrReplaceTempView("hms_eo_trfloss_config2")

      val point_code: Array[String] = hms_eo_trfloss_config_DF.map(x => x.getString(1)).collect()
      //val point_code = Array("MPED_ID_13_BM_P1R0")
      //println(point_code)
      //for(i <- 0 until  point_code.length){println(point_code(i))}
      //GetData.getNoLoop(ss,sc,hTable,point_code,time,step)
      GetData.getBySecond(ss, sc, hTable, point_code, time1, step, 900)
      ss.sql("select * from t_ods_hbase_edos_tmp").createOrReplaceTempView("t_ods")
      GetData.getBySecond(ss, sc, hTable, point_code, time2, step, 900)
      //ss.sql("select * from t_ods_hbase_edos_tmp").createOrReplaceTempView("t_ods2")

      //ss.sql("select * from t_ods_hbase_edos_tmp a join t_ods b on a.checkPoint=b.checkPoint order by  a.checkPoint,a.createTime ").show(200)
      //ss.sql("select a.checkPoint,a.createTime,a.pvalue  from t_ods a where a.createTime in ('2020/06/01 00:00:00')").show(500,0)
      //ss.sql("select a.checkPoint,a.createTime,a.pvalue  from t_ods_hbase_edos_tmp a where a.createTime in ('2020/07/01 00:00:00')").show(500,0)


      val etlDF = ss.sql("select a.checkPoint,b.createTime,a.pvalue-b.pvalue as pvalue  from t_ods_hbase_edos_tmp a join t_ods b on a.checkPoint=b.checkPoint and to_date(replace(a.createTime,'/','-'))=add_months(to_date(replace(b.createTime,'/','-')),1)")

      //etlDF.orderBy($"checkPoint", $"createTime").show(200, 0)

      etlDF.createOrReplaceTempView("t_ods_hbase_edos2")


      val exeString = "select createTime,sum(case when tlc_branch='daj_energy' then pvalue*org_id else 0 end)/100 daj_energy" +
        ",sum(case when tlc_branch='daj_grid_energy' then pvalue*org_id*is_positive else 0 end)/10000 daj_grid_energy " +
        ",sum(case when tlc_branch='daj_trf_energy' then pvalue*org_id else 0 end)/100 daj_trf_energy " +
        ",sum(case when tlc_branch='erj_energy' then pvalue*org_id*is_positive else 0 end)/100 erj_energy " +
        ",sum(case when tlc_branch='erj_grid_energy' then pvalue*org_id*is_positive else 0 end)/10000 erj_grid_energy " +
        ",sum(case when tlc_branch='erj_trf_energy' then pvalue*org_id else 0 end)/1000 erj_trf_energy " +
        ",sum(case when tlc_branch='erj_self_energy' then pvalue*org_id else 0 end)/100 erj_trf_loss " +
        ",sum(case when tlc_branch='erj_other_energy' then pvalue*org_id else 0 end)/100 qt_trf_loss " +
        "  from t_ods_hbase_edos2 a left join hms_eo_trfloss_config2 b on a.checkPoint=b.point_code  group by createTime"
      val tmp1resultDf = ss.sql(exeString).cache()

      tmp1resultDf.createOrReplaceTempView("tmp1resultDf")

      val tablename = "hms_eo_trfloss_analysis"

      ///////变损率 月报表substr(createTime,12,8)='00:00:00

      val resultDf3 = ss.sql("select '2' type,substr(createTime,1,7) time,daj_energy,erj_energy,daj_grid_energy,erj_grid_energy,daj_trf_energy,erj_trf_energy" +
        ",erj_trf_loss,qt_trf_loss,(daj_energy-daj_grid_energy-daj_trf_energy)/daj_energy as rate_trf_daj,(erj_energy-erj_grid_energy-erj_trf_energy-erj_trf_loss-qt_trf_loss)/erj_energy as rate_trf_erj" +
        ",(daj_energy+erj_energy-daj_grid_energy-daj_trf_energy-erj_grid_energy-erj_trf_energy-erj_trf_loss-qt_trf_loss)/(daj_energy+erj_energy) rate_trf_all" +
        ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time from tmp1resultDf where daj_energy<>0")
      resultDf3.show()
      val delString4 = s"delete from $tablename where time between substr('$lastDate',1,7) and substr('$lastDate',1,7) and type='2'"

      MysqlDml.delete2(delString4)
      //写mysql
      resultDf3.write
        .format("jdbc")
        .mode("append")
        .option("url", MysqlDml.url2)
        .option("dbtable", tablename)
        .option("user", MysqlDml.user2)
        .option("password", MysqlDml.password2)
        .option("driver", MysqlDml.driver2)
        .save()


      //////////////厂用电月
      val result2Df = ss.sql("select '2' type,substr(createTime,1,7) time,erj_energy,erj_trf_energy erj_fs_energy,daj_energy,daj_trf_energy daj_fs_energy,erj_energy+daj_energy all_energy" +
        ",erj_trf_energy+daj_trf_energy all_fs_energy" +
        ",(erj_trf_energy+daj_trf_energy)/(daj_energy+erj_energy) rate_fs" +
        ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time from tmp1resultDf where daj_energy<>0")
      val tablename2 = "hms_eo_fs_analysis"
      result2Df.show()
      val delString5 = s"delete from $tablename2 where time between substr('$lastDate',1,7) and substr('$lastDate',1,7) and type='2'"

      MysqlDml.delete2(delString5)
      //写mysql
      result2Df.write
        .format("jdbc")
        .mode("append")
        .option("url", MysqlDml.url2)
        .option("dbtable", tablename2)
        .option("user", MysqlDml.user2)
        .option("password", MysqlDml.password2)
        .option("driver", MysqlDml.driver2)
        .save()

    }
  }

  //@202107:未使用
  @Deprecated
  def calculateData_year(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int): Unit = {
    import ss.implicits._
    var startTime = time(0)
    val lastDate = time(1)
    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
    var endTimeStamp = sdf.parse(lastDate).getTime()
    var endTime = sdf.format(new Date(endTimeStamp + 3600 * 24 * 1000))
    println(endTime)
    var startTime2 = lastDate.substring(0, 5) + "01/01 00:00:00"
    val endTime2 = endTime + " 00:00:00"
    val time1 = Array(startTime2, startTime2)
    val time2 = Array(endTime2, endTime2)
    println(endTime.substring(5, 10))
    if (endTime.substring(5, 10) == "01/01") {
      val schema = new StructType()
        .add("point_name", types.StringType)
        .add("point_code", types.StringType)
        .add("tlc_branch", types.StringType)
        .add("multi_ratio", types.DoubleType)

      val hms_eo_trfloss_config_DF = ss.read.schema(schema).csv(MysqlDml.fileprefix + "/user/hive/trfloss.csv")
      hms_eo_trfloss_config_DF.createOrReplaceTempView("hms_eo_trfloss_config")

      ss.sql("select point_name,point_code,tlc_branch,multi_ratio org_id,case when instr(point_name,'励磁')>0 or instr(point_code,'P2R0')>0 then -1 else 1 end as is_positive from hms_eo_trfloss_config").createOrReplaceTempView("hms_eo_trfloss_config2")

      val point_code: Array[String] = hms_eo_trfloss_config_DF.map(x => x.getString(1)).collect()
      //val point_code = Array("MPED_ID_13_BM_P1R0")
      //println(point_code)
      //for(i <- 0 until  point_code.length){println(point_code(i))}
      //GetData.getNoLoop(ss,sc,hTable,point_code,time,step)
      GetData.getBySecond(ss, sc, hTable, point_code, time1, step, 900)
      ss.sql("select * from t_ods_hbase_edos_tmp").createOrReplaceTempView("t_ods")
      GetData.getBySecond(ss, sc, hTable, point_code, time2, step, 900)
      //ss.sql("select * from t_ods_hbase_edos_tmp").createOrReplaceTempView("t_ods2")
      //ss.sql("select * from t_ods_hbase_edos_tmp a join t_ods b on a.checkPoint=b.checkPoint order by  a.checkPoint,a.createTime ").show(200)
      //ss.sql("select * from t_ods2 a join t_ods b on a.checkPoint=b.checkPoint order by  a.checkPoint,a.createTime ").show(200)


      val etlDF = ss.sql("select a.checkPoint,b.createTime,a.pvalue-b.pvalue as pvalue  from t_ods_hbase_edos_tmp a join t_ods b on a.checkPoint=b.checkPoint and to_date(replace(a.createTime,'/','-'))=add_months(to_date(replace(b.createTime,'/','-')),12)")

      //etlDF.orderBy($"checkPoint", $"createTime").show(200, 0)

      etlDF.createOrReplaceTempView("t_ods_hbase_edos2")


      val exeString = "select createTime,sum(case when tlc_branch='daj_energy' then pvalue*org_id else 0 end)/100 daj_energy" +
        ",sum(case when tlc_branch='daj_grid_energy' then pvalue*org_id*is_positive else 0 end)/10000 daj_grid_energy " +
        ",sum(case when tlc_branch='daj_trf_energy' then pvalue*org_id else 0 end)/100 daj_trf_energy " +
        ",sum(case when tlc_branch='erj_energy' then pvalue*org_id*is_positive else 0 end)/100 erj_energy " +
        ",sum(case when tlc_branch='erj_grid_energy' then pvalue*org_id*is_positive else 0 end)/10000 erj_grid_energy " +
        ",sum(case when tlc_branch='erj_trf_energy' then pvalue*org_id else 0 end)/1000 erj_trf_energy " +
        ",sum(case when tlc_branch='erj_self_energy' then pvalue*org_id else 0 end)/100 erj_trf_loss " +
        ",sum(case when tlc_branch='erj_other_energy' then pvalue*org_id else 0 end)/100 qt_trf_loss " +
        "  from t_ods_hbase_edos2 a left join hms_eo_trfloss_config2 b on a.checkPoint=b.point_code  group by createTime"

      println(exeString)
      val tmp1resultDf = ss.sql(exeString).cache()

      tmp1resultDf.createOrReplaceTempView("tmp1resultDf")

      val tablename = "hms_eo_trfloss_analysis"

      ///////变损率 月报表substr(createTime,12,8)='00:00:00

      val resultDf3 = ss.sql("select '1' type,substr(createTime,1,4) time,daj_energy,erj_energy,daj_grid_energy,erj_grid_energy,daj_trf_energy,erj_trf_energy" +
        ",erj_trf_loss,qt_trf_loss,(daj_energy-daj_grid_energy-daj_trf_energy)/daj_energy as rate_trf_daj,(erj_energy-erj_grid_energy-erj_trf_energy-erj_trf_loss-qt_trf_loss)/erj_energy as rate_trf_erj" +
        ",(daj_energy+erj_energy-daj_grid_energy-daj_trf_energy-erj_grid_energy-erj_trf_energy-erj_trf_loss-qt_trf_loss)/(daj_energy+erj_energy) rate_trf_all" +
        ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time from tmp1resultDf where daj_energy<>0")
      resultDf3.show()
      val delString4 = s"delete from $tablename where time between substr('$lastDate',1,4) and substr('$lastDate',1,4) and type='1'"

      MysqlDml.delete2(delString4)
      //写mysql
      resultDf3.write
        .format("jdbc")
        .mode("append")
        .option("url", MysqlDml.url2)
        .option("dbtable", tablename)
        .option("user", MysqlDml.user2)
        .option("password", MysqlDml.password2)
        .option("driver", MysqlDml.driver2)
        .save()


      //////////////厂用电月
      val result2Df = ss.sql("select '1' type,substr(createTime,1,4) time,erj_energy,erj_trf_energy erj_fs_energy,daj_energy,daj_trf_energy daj_fs_energy,erj_energy+daj_energy all_energy" +
        ",erj_trf_energy+daj_trf_energy all_fs_energy" +
        ",(erj_trf_energy+daj_trf_energy)/(daj_energy+erj_energy) rate_fs" +
        ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time from tmp1resultDf where daj_energy<>0")
      val tablename2 = "hms_eo_fs_analysis"
      result2Df.show()
      val delString5 = s"delete from $tablename2 where time between substr('$lastDate',1,4) and substr('$lastDate',1,4) and type='1'"

      MysqlDml.delete2(delString5)
      //写mysql
      result2Df.write
        .format("jdbc")
        .mode("append")
        .option("url", MysqlDml.url2)
        .option("dbtable", tablename2)
        .option("user", MysqlDml.user2)
        .option("password", MysqlDml.password2)
        .option("driver", MysqlDml.driver2)
        .save()

    }
  }
}

